package com.urick.loadbanlance.dubbo;

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * Design Theories:
 * <a href=""></a>
 *
 * @author uRick 2021/12/13 18:17
 * Email: rickwork@163.com
 * Wechat: yy3076815642
 * QQ: 3076815642
 * @since 1.0
 */
public abstract class ConsistentHashLoadBalance extends AbstractLoadBalance {
    private final ConcurrentMap<String, ConsistentHashSelector<ServerInstance>> selectors = new ConcurrentHashMap<>();

    /**
     * 仅限于test
     *
     * @return
     */
    protected abstract String getMockSetArgument();

    @Override
    protected ServerInstance doSelect(List<ServerInstance> instanceList) {
        String key = instanceList.get(0).getUid();
        // using the hashcode of list to compute the hash only pay attention to the elements in the list
        //list元素一旦发生变化，则hashCode也会变化
        int invokersHashCode = instanceList.hashCode();
        ConsistentHashSelector<ServerInstance> selector = selectors.get(key);
        //实例发生变化，key也发生变化
        if (selector == null || selector.identityHashCode != invokersHashCode) {
            selectors.put(key, new ConsistentHashSelector<>(this, instanceList, 160, invokersHashCode));
            selector = selectors.get(key);
        }
        return selector.select();
    }



    private static final class ConsistentHashSelector<T> {

        private ConsistentHashLoadBalance balance;

        private final TreeMap<Long, ServerInstance> virtualInstance;

        private final int replicaNumber;

        private final int identityHashCode;

        ConsistentHashSelector(ConsistentHashLoadBalance balance, List<ServerInstance> instanceList, int hashNodes, int identityHashCode) {
            this.balance = balance;
            this.virtualInstance = new TreeMap<>();
            this.identityHashCode = identityHashCode;
            this.replicaNumber = hashNodes == 0 ? 160 : hashNodes;
            for (ServerInstance instance : instanceList) {
                String address = instance.getIp() + ":" + instance.getPort();
                for (int i = 0; i < replicaNumber / 4; i++) {
                    byte[] digest = getMD5().digest((address + i).getBytes(StandardCharsets.UTF_8));
                    for (int h = 0; h < 4; h++) {
                        long m = hash(digest, h);
                        virtualInstance.put(m, instance);
                    }
                }
            }
        }

        public ServerInstance select() {
            String key = balance.getMockSetArgument();
            byte[] digest = getMD5().digest(key.getBytes(StandardCharsets.UTF_8));
            return selectForKey(hash(digest, 0));
        }

        private ServerInstance selectForKey(long hash) {
            Map.Entry<Long, ServerInstance> entry = virtualInstance.ceilingEntry(hash);
            if (entry == null) {
                entry = virtualInstance.firstEntry();
            }
            return entry.getValue();
        }

        private long hash(byte[] digest, int number) {
            return (((long) (digest[3 + number * 4] & 0xFF) << 24)
                    | ((long) (digest[2 + number * 4] & 0xFF) << 16)
                    | ((long) (digest[1 + number * 4] & 0xFF) << 8)
                    | (digest[number * 4] & 0xFF))
                    & 0xFFFFFFFFL;
        }

        private MessageDigest getMD5() {
            try {
                return MessageDigest.getInstance("MD5");
            } catch (NoSuchAlgorithmException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public static void main(String[] args) throws NoSuchAlgorithmException {
        MessageDigest md5 = MessageDigest.getInstance("MD5");

        byte[] digest = md5.digest("1232".getBytes(StandardCharsets.UTF_8));

        System.err.println();

    }

}
